package transform;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/3/19 16:08
 * @Version: 1.0
 * @description:kafka自定义对象类型传输
 */
public class UserConsumerDemo {

    public static void main(String[] args) {

        //1:创建配置对象
        Properties properties = new Properties();

        //2：配置kafka集群的参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");

        //3：配置key的反序列化器
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);

        //4：配置value的反序列化器
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,UserDeSerializer.class);

        //设定消费组的名称
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"g2");

        //5：取消自动提交offerset偏移量
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        //6：设置自动规定时间提交偏移量
//        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,50000);

        //7：创建消费者对象
        KafkaConsumer<Integer, User> userKafkaConsumer = new KafkaConsumer<Integer, User>(properties);

        //8：定阅主题
        userKafkaConsumer.subscribe(Arrays.asList("t4"));

        //9：获取主题的内容信息
        while (true){
            //设置拉取新数据的，时间限制
            ConsumerRecords<Integer, User> record = userKafkaConsumer.poll(Duration.ofSeconds(5));
            for (ConsumerRecord<Integer, User> records : record) {
                System.out.println(records.key()+"|"+records.value().getName()+"|"+records.offset());

            }
        }


    }
}
